package com.haiyou.data.common.accessor.rsync;

import java.io.Serializable;
import java.time.Duration;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.Validate;
import org.redisson.api.RAtomicLong;
import org.redisson.api.RMapCache;

import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.haiyou.common.constants.ServerDataConstants;
import com.haiyou.common.entity.Entity;
import com.haiyou.common.spring.schedule.ActorScheduledExecutor;
import com.haiyou.common.system.Properties;
import com.haiyou.common.thread.ThreadManager;
import com.haiyou.common.thread.ThreadManager.ThreadEnum;
import com.haiyou.common.thread.task.AbstractOrderedTask;
import com.haiyou.common.util.collect.CollectionUtils;
import com.haiyou.common.util.network.AddressUtils;
import com.haiyou.common.util.random.ProbabilityUtils;
import com.haiyou.common.util.reflect.AnnotationFinder;
import com.haiyou.data.common.accessor.AbstractAccessor;
import com.haiyou.data.common.accessor.Accessor;
import com.haiyou.data.common.accessor.annotation.CachedPolicy;
import com.haiyou.data.common.accessor.annotation.ExpiryPolicy;
import com.haiyou.data.common.accessor.annotation.SizePolicy;
import com.haiyou.data.common.accessor.caffeine.CaffeineAccessor;
import com.haiyou.data.common.accessor.redis.redisson.RedissonAccessor;
import com.haiyou.data.common.cache.redis.RKeys;
import com.haiyou.data.common.cache.redis.key.RedisPrefix;
import com.haiyou.data.common.persister.mysql.annotation.SyncPolicy;
import com.haiyou.data.common.persister.mysql.persister.Persister;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class Synchronizer<K extends Serializable & Comparable<K>, V extends Entity<K>> implements Flusher<K, V> {

	private static final class Lazy {
		private static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE;
		static {
			int nThreads = ThreadManager.getThreadCount(ThreadEnum.data_flush_thread_num);
			SCHEDULED_EXECUTOR_SERVICE = new ActorScheduledExecutor(nThreads, Synchronizer.class.getSimpleName());
		}
	}

	public static ScheduledExecutorService scheduler() {
		return Lazy.SCHEDULED_EXECUTOR_SERVICE;
	}

	private final Accessor<K, V> accessor;
	private final Persister<K, V> persister;
	private final Map<K, V> syncMap;

	@Getter
	private int initialCapacity;
	@Getter
	private int maximumSize;

	@Getter
	private int timeToLive;
	@Getter
	private TimeUnit timeToLiveUnit;

	private TimeUnit delayUnit;
	private int initialDelay;
	private int delay;
	
	/**
	 * 是否需要执行数据库操作
	 */
	private boolean flushDB = true;

	private ScheduledFuture<?> scheduledFuture;

	public Synchronizer(AbstractAccessor<K, V> accessor, Persister<K, V> persister) {
		super();
		this.accessor = accessor;
		this.persister = persister;
		if (accessor instanceof CaffeineAccessor) {
			syncMap = Maps.newConcurrentMap();
		} else if (accessor instanceof RedissonAccessor) {
			RedissonAccessor<?, ?> redissonAccessor = (RedissonAccessor<?, ?>) accessor;
			syncMap = redissonAccessor.getRedissonClient().getMapCache(RKeys.createRKey(RMapCache.class,
					redissonAccessor.getKeyPrefix(), getClass().getSimpleName()));
		} else {
			throw new NullPointerException(accessor + "");
		}
	}

	public void init() {

		//判断是否需要刷入DB
		this.flushDB = ServerDataConstants.contains(this.persister.getEntityClass());
		
		if(!Properties.synchronizer_flush_open()) {
			flushDB = false; 
		}
		
		// 缓存策略
		CachedPolicy cachedPolicy = AnnotationFinder.findAnnotation(accessor.getInitClass(), CachedPolicy.class);

		// 缓存大小
		SizePolicy sizePolicy = cachedPolicy.sizePolicy();
		initialCapacity = sizePolicy.initialCapacity();
		Validate.isTrue(initialCapacity > 0, "%s 缓存初始容量非法：%s", accessor, initialCapacity);
		maximumSize = sizePolicy.maximumSize();

		// 缓存有效时长
		ExpiryPolicy expiryPolicy = cachedPolicy.expiryPolicy();
		timeToLiveUnit = expiryPolicy.timeUnit();
		Validate.isTrue(expiryPolicy.timeToLive() == -1 || expiryPolicy.timeToLive() > 0);
		if (expiryPolicy.timeToLive() > 0)
			timeToLive = ProbabilityUtils.xTo2x(expiryPolicy.timeToLive());

		// 异步定时入库时效
		SyncPolicy syncPolicy = cachedPolicy.syncPolicy();
		delayUnit = syncPolicy.timeUnit();
		if (syncPolicy.delay() > 0) {
			initialDelay = syncPolicy.initialDelay() > 0 ? ProbabilityUtils.xTo2x(syncPolicy.initialDelay()) : 0;
			delay = ProbabilityUtils.xTo2x(syncPolicy.delay());
			delay = Math.max(1, delay);

			// 保证缓存过期之前一定要入库，有效时长必须大于延迟入库时间
			if (timeToLive > 0)
				Validate.isTrue(
						timeToLiveUnit.toMillis(timeToLive) > delayUnit.toMillis(initialDelay)
								+ delayUnit.toMillis(delay * 2),
						"[%s]过期时间[%d]太小，同步时间间隔过大[%d]", accessor.getValueClass(), timeToLive + "," + timeToLiveUnit,
						delay + "," + delayUnit);
				
				//对应节点启动对应定时器
				if(flushDB) {
					// 开启定时入库时效任务
					scheduledFuture = scheduler().scheduleAtFixedRate(new PersistTask(false), initialDelay, delay, delayUnit);
					log.info("{} 启动更新定时器, initialDelay={},delay={}", accessor.getValueClass().getName(), initialDelay, delay);
				}else {
					//不启动
				}
			
			} else {

			log.info("{} no scheduled, save now ,time to Live long long.", accessor);

		}

	}

	public void destory() {
		
		if(!this.flushDB) {
			return ;
		}
		
		scheduler().execute(new PersistTask(true));
	}
	
	public void flushAll() {
		if(!this.flushDB) {
			return ;
		}
		scheduler().execute(new PersistTask(false));
	}
	

	@Override
	public void giveUp(Iterable<? extends K> keys) {
		K[] keyArr = Iterables.toArray(keys, accessor.getKeyClass());
		if (syncMap instanceof RMapCache) {
			((RMapCache<K, V>) syncMap).fastRemoveAsync(keyArr);
		} else {
			syncMap.keySet().removeAll(Sets.newHashSet(keyArr));
		}
		checkFlushingNow();
	}

	@Override
	public void flush(Map<? extends K, ? extends V> map) {
		if (syncMap instanceof RMapCache)
			((RMapCache<K, V>) syncMap).putAllAsync(map, persister.sqlBatchNum());
		else
			syncMap.putAll(map);
		checkFlushingNow();
	}

	private void checkFlushingNow() {
		//对应节点更新对应数据
		if(!this.flushDB) {
			return ;
		}

		if (scheduledFuture == null || persister.sqlBatchNum() <= syncMap.size()) {
			scheduler().execute(new PersistTask(false));
		}
	}

	private final class PersistTask extends AbstractOrderedTask<Class<V>> {

		private final boolean last;
		
		private PersistTask(boolean last) {
			super(persister.getEntityClass());
			this.last = last;
		}

		@Override
		protected void exec() throws Exception {
			if (accessor instanceof RedissonAccessor) {
				
				RedissonAccessor<?, ?> redissonAccessor = (RedissonAccessor<?, ?>) accessor;
			
				RAtomicLong atomicExecutor = redissonAccessor.getRedissonClient().getAtomicLong(
						RKeys.createRKey(RAtomicLong.class, redissonAccessor.getKeyPrefix(), "Atomic-Flusher"));
				// 是否mail相关的更新
				boolean needCheckFinish = needCheck(redissonAccessor.getKeyPrefix());
				try {
					if (atomicExecutor.compareAndSet(0L, 1L)) {
						// 如果是邮件更新 设置过期时间
						if(needCheckFinish) {
							needCheckFinish = false;
							atomicExecutor.expire(Duration.ofMillis(30000));
						}
						
						if(log.isDebugEnabled()) {
							log.debug("获取到同步权限[{}]", AddressUtils.getHostAddress());
						}
						flushing();
					} else {
						if(log.isDebugEnabled()) {
							log.debug("未获取到同步权限[{}]", AddressUtils.getHostAddress());
						}
					}
				} finally {
					if (!needCheckFinish) {
						atomicExecutor.compareAndSet(atomicExecutor.get(), 0L);
					}
				}
			} else {
				flushing();
			}
		}

		private boolean needCheck(String keyPrefix) {
			return RedisPrefix.PLAYER_MAIL.equals(keyPrefix) || RedisPrefix.PLAYER_MAIL_INDEX.equals(keyPrefix);
		}

		private void flushing() {
			if(log.isDebugEnabled()) {
				log.debug("flushing to db [{}:{}]", persister.getEntityClass(), syncMap.size());
			}
			Iterator<Entry<K, V>> iterator = syncMap.entrySet().iterator();
			Set<V> saveValues = null;
			while (iterator.hasNext()) {
				if (saveValues == null) {
					saveValues = Sets.newHashSet();
				}
				Entry<K, V> entry = iterator.next();
				V value = syncMap.remove(entry.getKey());
				if(value == null) {
					continue;
				}
				saveValues.add(value);
				
				if (saveValues.size() >= persister.sqlBatchNum() || !iterator.hasNext()) {
					//Stopwatch stopwatch = Stopwatch.createStarted();
					persister.insertOrUpdate(saveValues);
					//log.info("{} 运行时间:{}",this.getClass().getSimpleName(),stopwatch.elapsed(TimeUnit.MILLISECONDS));
					saveValues = null;
				}
			}
			if (last && !CollectionUtils.isEmpty(syncMap))
				flushing();
		}
	}

}
